package com.fenquen.dfs.router.rocksdb;

import com.alibaba.fastjson.JSON;
import com.fenquen.dfs.router.config.DFSRouterConfig;
import com.fenquen.dfs.router.exception.NoStorageFoundException;
import com.fenquen.dfs.router.exception.NoAvailableStorageException;
import com.fenquen.dfs.base.protocol.data.storage_source.req_from_storage.StorageNode;
import com.fenquen.dfs.base.protocol.data.storage_source.req_from_storage.Report;
import org.rocksdb.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@Component
public class RocksDBOperator {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocksDBOperator.class);

    @Autowired
    private RocksDB rocksDB;

    @Autowired
    private RocksDBProp rocksDBProp;

    private byte[] currentKey;

    @Autowired
    private DFSRouterConfig dfsRouterConfig;


    public List<StorageNode> getAvailableStorageList(Long uploadFileSize) throws NoAvailableStorageException {
        List<StorageNode> targetStorageList = new ArrayList<>(dfsRouterConfig.replicaNum);

        // 怎么做到随机均匀的获取存储节点防止倾斜？
        RocksIterator rocksIterator = rocksDB.newIterator(rocksDBProp.tableName_columnFamilyHandle.get("STORAGE_NODE"));

        if (currentKey == null) {
            rocksIterator.seekToFirst();
        } else {
            rocksIterator.seek(currentKey);
        }

        // 遍历寻找的次数
        int iterateCount = 0;

        for (; rocksIterator.isValid(); rocksIterator.next()) {
            // 已找到足够多的replica数量
            if (targetStorageList.size() >= dfsRouterConfig.replicaNum) {
                return targetStorageList;
            }

            // byte[]对应的是json文本
            StorageNode storageNode = JSON.parseObject(new String(rocksIterator.value()), StorageNode.class);
            currentKey = rocksIterator.key();


            iterateCount++;

            // 该存储节点满足条件
            if (storageNode.remainingSpaceByte > uploadFileSize &&
                    dfsRouterConfig.maxSubmitInterval > System.currentTimeMillis() - storageNode.timestamp) {
                if (!targetStorageList.contains(storageNode)) {
                    targetStorageList.add(storageNode);
                }
            }

        }

        // 要是趟数不够还要在从头
        if (iterateCount >= dfsRouterConfig.replicaNum * 16) {
            if (targetStorageList.isEmpty()) {
                throw new NoAvailableStorageException();
            }
            return targetStorageList;
        }

        // 遍历的次数还是不够,要再从头起始
        for (rocksIterator.seekToFirst(); rocksIterator.isValid(); rocksIterator.next()) {
            if (targetStorageList.size() >= dfsRouterConfig.replicaNum) {
                return targetStorageList;
            }

            // byte[]对应的是json文本
            StorageNode storageNode = JSON.parseObject(new String(rocksIterator.value()), StorageNode.class);
            currentKey = rocksIterator.key();

            iterateCount++;

            // 该存储节点满足条件
            if (storageNode.remainingSpaceByte > uploadFileSize &&
                    dfsRouterConfig.maxSubmitInterval > System.currentTimeMillis() - storageNode.timestamp) {
                if (!targetStorageList.contains(storageNode)) {
                    targetStorageList.add(storageNode);
                }
            }
        }

        // 要是没有知道存储节点直接异常
        if (targetStorageList.isEmpty()) {
            throw new NoAvailableStorageException();
        }

        return targetStorageList;
    }

    /**
     * 保存各存储节点定时上报的节点信息
     *
     * @param storageNode
     * @throws RocksDBException
     */
    public void recordStorageNode(StorageNode storageNode) throws RocksDBException {
        // 获取对应的表的columnFamilyHandle
        ColumnFamilyHandle columnFamilyHandle = rocksDBProp.tableName_columnFamilyHandle.get("STORAGE_NODE");
        // addr_节点
        rocksDB.put(columnFamilyHandle,
                storageNode.getAddr().getBytes(StandardCharsets.UTF_8), JSON.toJSONString(storageNode).getBytes(StandardCharsets.UTF_8));

        // RocksIterator rocksIterator = rocksDB.newIterator(new ReadOptions().setPrefixSameAsStart(true));
    }

    /**
     * 处理存储节点在删除或保存文件之后上报的回执
     *
     * @param report
     * @throws RocksDBException
     */
    public void recordReport(Report report) throws RocksDBException {
        // 该存储节点保存文件失败
        if (report.success == Report.Success.失败) {
            LOGGER.error(report.msg);
            return;
        }

        ColumnFamilyHandle columnFamilyHandle = rocksDBProp.tableName_columnFamilyHandle.get("FILE_RECORD");
        String key = report.uploadFileUUID + report.getAddr();

        // 删掉文件
        if (report.flag == Report.Flag.删除) {
            rocksDB.delete(columnFamilyHandle, key.getBytes(StandardCharsets.UTF_8));
        } else { // 添加文件
            rocksDB.put(columnFamilyHandle,
                    key.getBytes(StandardCharsets.UTF_8), JSON.toJSONString(report).getBytes(StandardCharsets.UTF_8));
        }
    }

    public List<Report> getReportList(String uploadFileUUID) throws NoStorageFoundException {
        ColumnFamilyHandle columnFamilyHandle = rocksDBProp.tableName_columnFamilyHandle.get("FILE_RECORD");

        RocksIterator rocksIterator = rocksDB.newIterator(columnFamilyHandle,
                new ReadOptions().setPrefixSameAsStart(true));

        rocksIterator.seek(uploadFileUUID.getBytes(StandardCharsets.UTF_8));

        List<Report> reportList = new ArrayList<>();
        for (; rocksIterator.isValid(); rocksIterator.next()) {
            reportList.add(JSON.parseObject(new String(rocksIterator.value()), Report.class));
        }

        if (reportList.isEmpty()) {
            throw new NoStorageFoundException("can not locate the storage node for the uploadFileUUID " + uploadFileUUID);
        }

        return reportList;
    }
}
